package com.enjoy.rocketmq.simple;

import com.enjoy.rocketmq.Util;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

public class LitePullConsumerAssign {
    public static void main(String[] args) throws Exception {
        Util.loadProperties();

        DefaultLitePullConsumer litePullConsumer = new DefaultLitePullConsumer("lite_pull_consumer");
        litePullConsumer.setNamesrvAddr(System.getProperty("nameserver.addr"));

        litePullConsumer.setAutoCommit(false);
        litePullConsumer.start();

        Collection<MessageQueue> messageQueues = litePullConsumer.fetchMessageQueues("TopicTest");

        List<MessageQueue> list = new ArrayList<>(messageQueues);


        List<MessageQueue> assignList = new ArrayList<>();

        for (int i = 0; i < list.size() / 2; i++) {
            assignList.add(list.get(i));
        }

        litePullConsumer.assign(assignList);

        litePullConsumer.seek(assignList.get(0), 10);

        while (true){
            List<MessageExt> messageExts = litePullConsumer.poll();

            System.out.printf("%s%n", messageExts);

            litePullConsumer.commitSync();
        }


    }
}
